package com.we.flink.function;

import org.apache.flink.api.common.functions.MapFunction;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.we.flink.bean.FlinkCheckpointInfo;
import com.we.utils.HttpUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.text.SimpleDateFormat;

/**
 * @author ：Jason
 * @date ：Created in 2023/5/30 4:23 PM
 * @description： 通过 Flink Rest API 获取任务的 job_id,job_name,state,checkpoint path 信息
 * @version: 1.0
 */
public class MapFlinkRestAPIFunction
        implements MapFunction<FlinkCheckpointInfo, FlinkCheckpointInfo> {
    private static final Logger LOG = LoggerFactory.getLogger(HttpUtils.class);
    private static final String FLINK_JOB_OVERVIEW =
            "http://rtmdw01:8088/proxy/${application_id}/v1/jobs/overview";
    private static final String FLINK_CHECKPOINT =
            "http://rtmdw01:8088/proxy/${application_id}/v1/jobs/${job_id}/checkpoints";
    private static final SimpleDateFormat SDF = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    @Override
    public FlinkCheckpointInfo map(FlinkCheckpointInfo value) {
        final String overviewUrl =
                FLINK_JOB_OVERVIEW.replace("${application_id}", value.getApplicationId());
        try {
            // 获取 job_id,job_name,state 信息
            final String overview = HttpUtils.httpGet(overviewUrl);
            final JSONObject jsonObject = JSON.parseObject(overview);
            final JSONArray jobs = jsonObject.getJSONArray("jobs");
            if (!jobs.isEmpty()) {
                final JSONObject job = (JSONObject) jobs.get(0);
                final String jid = job.getString("jid");
                value.setJobId(jid);
                value.setJobName(job.getString("name"));
                value.setJobState(job.getString("state"));
                final String checkpointUrl =
                        FLINK_CHECKPOINT
                                .replace("${application_id}", value.getApplicationId())
                                .replace("${job_id}", jid);
                // 获取 checkpoint 路径信息
                final String checkpoint = HttpUtils.httpGet(checkpointUrl);
                final JSONObject jsonObject1 = JSON.parseObject(checkpoint);
                final JSONObject completed =
                        jsonObject1.getJSONObject("latest").getJSONObject("completed");
                if (completed != null) {
                    value.setExternalPath(completed.getString("external_path"));
                }
                value.setUpdateTime(SDF.format(System.currentTimeMillis()));
            }
        } catch (IOException e) {
            LOG.info("get checkpoint failed, {}, data: {}", e.getMessage(), value);
            e.printStackTrace();
        }
        return value;
    }
}
